package com.ljh.seckill.mq;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.common.RemotingUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

/**
 * @Author grung
 * @Date 2021/2/23 13:48
 * @Version 1.0
 */

@Component
public class Producer {

    private Logger logger= LoggerFactory.getLogger(Producer.class);

    @Value("${apache.rocketmq.namesrvAddr")
    private String namesrvAddr;

    @Value("${apache.rocketmq.producer.groupName}")
    private String groupName;

    private DefaultMQProducer producer;


    /**
     * 启动生产者
     * Constructor(构造方法) -> @Autowired(依赖注入) -> @PostConstruct(注释的方法)
     */
    @PostConstruct
    public void start() {

        logger.info("开始启动RocketMQ生产者 groupName:{}", groupName);
        producer = new DefaultMQProducer(groupName);
        producer.setNamesrvAddr(namesrvAddr);

        producer.setClientIP(RemotingUtil.getLocalAddress());
        //是否启用vip netty通道以发送消息,默认值：-D com.rocketmq.sendMessageWithVIPChannel参数的值，若无则是false
        producer.setVipChannelEnabled(false);
        //允许发送的最大消息体大小，默认值 1024 * 1024 * 4;  4M
        producer.setMaxMessageSize(1024 * 1024 * 4);
        //同步消息发送失败重试次数，默认2次
        producer.setRetryTimesWhenSendFailed(10);
        //异步消息发送失败重试次数，默认2次
        producer.setRetryTimesWhenSendAsyncFailed(10);
        producer.setRetryAnotherBrokerWhenNotStoreOK(true);
        producer.setDefaultTopicQueueNums(1);
        try {
            producer.start();
            logger.info("RocketMQ生产者启动成功");
        } catch (MQClientException e) {
            logger.info("RocketMQ生产者启动失败");
        }
    }

    /**
     * 关闭生产者
     */
    @PreDestroy
    public void shutdown() {
        logger.info("开始关闭RocketMQ生产者");
        producer.shutdown();
        logger.info("RocketMQ生产者关闭成功");
    }

    /**
     * 发送消息
     *
     * @param topic   Message所属的Topic
     * @param tag     Message Tag 可理解为Gmail中的标签，对消息进行再归类，方便Consumer指定过滤条件在MQ服务器过滤
     * @param content 消息内容
     */
    public Boolean sendMsg(String topic, String tag, String content) {
        try {
            Message messageVo = new Message(topic, tag, content, content.getBytes(RemotingHelper.DEFAULT_CHARSET));
            SendResult result = producer.send(messageVo);

            if (null != result && result.getSendStatus().equals(SendStatus.SEND_OK)) {
                logger.info(result.toString());
                return true;
            }
            return false;
        } catch (Exception e) {
            logger.error("rocketMq消息发送异常:{}",
                    new StringBuilder("topic:").append(topic)
                            .append("<br/>tag:").append(tag)
                            .append("<br/>body:").append(content).toString(), e);

            return false;
        }
    }

    /**
     * 发送消息 自定义超时时间
     *
     * @param topic   Message所属的Topic
     * @param tag     Message Tag 可理解为Gmail中的标签，对消息进行再归类，方便Consumer指定过滤条件在MQ服务器过滤
     * @param content 消息内容
     * @param timeout Timeout for sending messages. 默认时间是 3000ms
     */
    public Boolean sendMsg(String topic, String tag, String content, Long timeout) {
        try {
            Message messageVo = new Message(topic, tag, content, content.getBytes(RemotingHelper.DEFAULT_CHARSET));
            SendResult result = producer.send(messageVo, timeout);
            if (null != result && result.getSendStatus().equals(SendStatus.SEND_OK)) {
                logger.info(result.toString());
                return true;
            }
            return false;
        } catch (Exception e) {
            logger.error("rocketMq消息发送异常:{}",
                    new StringBuilder("topic:").append(topic)
                            .append("<br/>tag:").append(tag)
                            .append("<br/>body:").append(content).toString(), e);
            return false;
        }
    }

    /**
     * 发送延时消息
     *
     * @param topic
     * @param tag
     * @param content
     * @param delayTimeLevel 延迟级别
     */
    public Boolean sendDelayMsg(String topic, String tag, String content, DelayTimeLevel delayTimeLevel) {
        try {
            Message messageVo = new Message(topic, tag, content, content.getBytes(RemotingHelper.DEFAULT_CHARSET));
            messageVo.setDelayTimeLevel(delayTimeLevel.getLevel());
            SendResult result = producer.send(messageVo);

            if (null != result && result.getSendStatus().equals(SendStatus.SEND_OK)) {
                logger.info(result.toString());
                return true;
            }
            return false;
        } catch (Exception e) {
            logger.error("rocketMq消息发送异常:{}",
                    new StringBuilder("topic:").append(topic)
                            .append("<br/>tag:").append(tag)
                            .append("<br/>body:").append(content).toString(), e);

            return false;
        }
    }
}
